package com.ruibang.glass.job.service;

import com.ruibang.glass.job.entity.DefectDataSource;
import com.ruibang.glass.job.entity.RQDataSource;
import com.ruibang.glass.job.entity.TimeInterval;
import com.ruibang.glass.job.feign.QualityApi;
import com.ruibang.glass.job.utils.JdbcOperUtils;
import com.teaming.cloud.framework2.common.model.ResultBody;
import com.teaming.cloud.framework2.common.tools.JsonUtils;
import com.teaming.cloud.framework2.elastic.job3.annotation.ElasticJobScheduler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.*;
import java.net.InetAddress;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;


/**
 * 基于注解形式script job
 * 抽取质量数据job
 */
@Service
//@ElasticJobScheduler(cron = "20 5 0 * * ?", shardingTotalCount = 1, name = "RQDataElasticJob",
//        shardingItemParameters = "0=0", jobParameters = "parameter")
@Slf4j
public class RQDataElasticJob implements SimpleJob {


    @Autowired
    private QualityApi qualityApi;

    @Autowired
    private AsyncTotalDataService asyncTotalDataService;


    @Override
    public void execute(ShardingContext shardingContext) {
        try {
            log.error(String.format("RQDataElasticJob starting..., IP: %s,Thread ID: %s", InetAddress.getLocalHost().getHostAddress(),
                    Thread.currentThread().getId()));
            Map<String, String> timeStamp = explainProperties();
            TimeInterval timeInterval = new TimeInterval();
            //通过lighting导入
            if (StringUtils.isBlank(timeStamp.get("inittime"))){
                String ctime = timeStamp.get("lasttime");
                String initTime = JdbcOperUtils.getTenMinsTimesBefore(ctime);
                timeStamp.put("inittime",initTime);
                saveProperties("inittime",initTime);
            }
                timeInterval.setStartTime(timeStamp.get("inittime"));
                timeInterval.setEndTime(JdbcOperUtils.getLatelyTenMinsTimes());
                String [] command = new String[]{"python", "/soft/pyscript/main.py", "-t", "rqd"};
                ProcessBuilder processBuilder = new ProcessBuilder(command);
                processBuilder.directory(new File("/soft/pyscript"));
                processBuilder.redirectErrorStream(true);
                processBuilder.redirectOutput(ProcessBuilder.Redirect.appendTo(new File("/soft/pyscript/log/rqd.log")));
                Process process = processBuilder.start();
                // 启动线程读取输出流
                BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
                new Thread(() -> {
                    String line;
                    try {
                        while ((line = reader.readLine()) != null) {
                            log.info(line); // 处理输出
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }).start();
                int exitCode = process.waitFor();
                if (reader != null) {
                    reader.close();
                }
                if (process != null) {
                    process.destroyForcibly();
                }
//            } else {
//                //通过insert写入，避免对业务造成影响
//                timeInterval.setStartTime(timeStamp.get("lasttime"));
//                timeInterval.setEndTime(JdbcOperUtils.getLatelyTenMinsTimes());
//                //修改开始时间到前20分钟
//                saveProperties(timeInterval.getEndTime());
//
//                Map<String, DefectDataSource> dataSourceMap = rqDataSource.getDataSources();
//                List<String> resultList = batchGetUserInfoByCompletableFuture(dataSourceMap,timeInterval);
//                resultList.forEach(System.out::println);
//                //修改时间到当前
//                saveProperties(timeInterval.getEndTime());
//            }
            ResultBody resultBody = qualityApi.defectHourTask(timeInterval);
            log.error(String.format("======RQDataElasticJob execute finish,resultBody:%s.", JsonUtils.toJsonString(resultBody)));
        }catch (Exception e){
            log.error(e.getMessage());
        }


    }

    List<String> batchGetUserInfoByCompletableFuture(Map<String, DefectDataSource> productionList,TimeInterval timeInterval) {
        List<CompletableFuture<String>>  prodInfoFutrues = productionList.entrySet().stream().map(e->asyncTotalDataService
                .asyncTotalData(e.getKey(), e.getValue(),timeInterval)).collect(Collectors.toList());
        return prodInfoFutrues.stream().map(CompletableFuture::join).
                collect(Collectors.toList());
    }


    private Map<String,String> explainProperties(){
        Properties prop = new Properties();
        Map<String,String> values = new HashMap<>();
        try (InputStream input = new FileInputStream("/soft/pyscript/config/timeCursor.ini")) {
            // 加载properties文件
            prop.load(input);
            // 假设属性文件中的键和值有特定格式，例如 [类型]
            // 解析这些键和值
            for (String key : prop.stringPropertyNames()) {
                String value = prop.getProperty(key);
                values.put(key,value);
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
        return values;
    }

    private void saveProperties(String pos,String timePoint){
        Properties prop = new Properties();
        try {
            InputStream input = new FileInputStream("/soft/pyscript/config/timeCursor.ini");
            // 加载properties文件
            prop.load(input);
            // 假设属性文件中的键和值有特定格式，例如 [类型]
            // 解析这些键和值
            prop.setProperty(pos, timePoint);
            OutputStream fos = new FileOutputStream("/soft/pyscript/config/timeCursor.ini");
            BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fos, StandardCharsets.UTF_8));
            bw.newLine();
            for(Enumeration<?> e = prop.keys(); e.hasMoreElements();) {
                String key = (String)e.nextElement();
                String val = prop.getProperty(key);
                bw.write(key + "=" + val);
                bw.newLine();
            }
            bw.flush();
            fos.close();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}
